package com.bw.flinkstreaming.state2.job1;

import org.apache.flink.api.common.functions.RichFlatMapFunction;
import org.apache.flink.api.common.state.ValueState;
import org.apache.flink.api.common.state.ValueStateDescriptor;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.util.Collector;

/**
 *
 * 需求：相同key的单词出现三次就计算平均值
 *   hadoop
 *   hive
 *   hive
 *   hadoop
 *   spark
 *   hadoop  -> 触发计算
 *   hive    -> 触发计算
 *   hadoop
 *   hadoop
 *   hadoop  -> 触发计算
 *
 *  自定义状态实现该需求，使用ValueState状态
 * */
public class ValueStateCountAvg extends RichFlatMapFunction<Tuple2<Long,Long>,Tuple2<Long,Double>> {

    //定义状态
    private ValueState<Tuple2<Long,Long>> countAvg;

    /**
     * 执行的时候会调用一次，在这里做状态初始化
     * */
    @Override
    public void open(Configuration parameters) throws Exception {
        //初始化状态的代码
        ValueStateDescriptor valueStateDescriptor = new ValueStateDescriptor<Tuple2<Long,Long>>(
                "avg",  //状态名称
                Types.TUPLE(Types.LONG,Types.LONG));    //状态存储的数据类型
        countAvg = getRuntimeContext().getState(valueStateDescriptor);  //初始化
    }

    /**
     * value：100，20
     * */
    @Override
    public void flatMap(Tuple2<Long, Long> value, Collector<Tuple2<Long, Double>> out) throws Exception {
        //获取当前的状态数据
        Tuple2<Long, Long> currentState = countAvg.value();

        if (currentState == null) {
            currentState = Tuple2.of(0L,0L);
        }

        //该key出现的次数
        currentState.f0 +=1;

        //更新状态的值
        currentState.f1 += value.f1;

        //更新状态的值
        countAvg.update(currentState);

        //判断是否满足3次
        if (currentState.f0 == 3) {
            double avg = (double)currentState.f1 / currentState.f0;
            out.collect(Tuple2.of(value.f0,avg));
            //清空状态
            countAvg.clear();
        }
    }
}
